<!DOCTYPE html>
<html lang="zh-CN">
<head>
        <meta http-equiv="content-type" content="text/html; charset=utf-8">
    <meta content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=0" name="viewport">
    <title>
        
            一个介于wait-free 和 lock-free的高性能MPSC队列
        
    </title>
    <link rel="icon" href="/img/favicon.png"/>
    
<link rel="stylesheet" href="/css/style.css">

    
<link rel="stylesheet" href="/css/font-awesome.min.css">

    <!-- 
<link rel="stylesheet" href="/css/gitment.css">
 -->
    
<link rel="stylesheet" href="/css/hljs.min.css">

    
<script src="/js/hljs.min.js"></script>
  
    <script>
        var _hmt = _hmt || [];
        (function() {
          var hm = document.createElement("script");
          hm.src = "https://hm.baidu.com/hm.js?8ad9e7d37411eee585b4bd7194806394";
          var s = document.getElementsByTagName("script")[0]; 
          s.parentNode.insertBefore(hm, s);
        })();
    </script>  
    <!-- 
<script src="/js/gitment.browser.js"></script>
   -->
    <link rel="stylesheet" href="https://unpkg.com/gitalk/dist/gitalk.css">
    <script src="https://unpkg.com/gitalk/dist/gitalk.min.js"></script>
<meta name="generator" content="Hexo 5.3.0"><link rel="alternate" href="/atom.xml" title="Nick" type="application/atom+xml">
</head>
<body>
    <header class="header" id="header">
    <h1>
        <a class="title" href="/">
            Nick
        </a>
    </h1>
    <h2>
        <a class="motto">
            天地间的怪事莫过于侦察别人的一些和自己绝不相干的事了
        </a>
    </h2>
    <nav class="navbar">
        <ul class="menu">
            
            
                <li class="menu-item">
                    <a href="/" class="menu-item-link">
                        首页
                    </a>
                </li>
            
                
            
                <li class="menu-item">
                    <a href="/archives" class="menu-item-link">
                        归档
                    </a>
                </li>
            
                
            
                <li class="menu-item">
                    <a target="_blank" rel="noopener" href="https://github.com/StoneQI" class="menu-item-link">
                        Github
                    </a>
                </li>
            
                
            
                <li class="menu-item">
                    <a href="/atom.xml" class="menu-item-link">
                        RSS
                    </a>
                </li>
            
                
            
                
            
                <li class="menu-item">
                    <a href="/category-archive" class="menu-item-link">
                        目录
                    </a>
                </li>
            
                
                
                <li class="menu-item">
                    <a class="menu-item-link search">
                        Search                   
                        <i class="fa fa-long-arrow-right search-icon" aria-hidden="true"></i>
                    </a>
                        <input placeholder="Search..." class="search-input" style="display:none;border:none!important;" onkeydown="onEnter(event)" onkeypress="onEnter(event)"></input>
                </li>
                
        </ul>
    </nav>
</header>
    <main class="main">
        <article class="post">
            <h1>
                <a class="title" href="/2021/03/04/go-mpsc-md/"> 
                    一个介于wait-free 和 lock-free的高性能MPSC队列 
                </a>
            </h1>
            <div class="meta">
                <a class="date"> 
                    <i class="fa fa-calendar" aria-hidden="true"></i>                    
                    2021-03-04   
                </a>
                
                <a class="category">
                    <i class="fa fa-th" aria-hidden="true"></i>  
                </a>
               
                <ul class="category-list"><li class="category-list-item"><a class="category-list-link" href="/categories/go/">go</a></li></ul>
                
                <a class="tag">
                    <i class="fa fa-tags" aria-hidden="true"></i>  
                </a>
                
                    <ul class="tag-list" itemprop="keywords"><li class="tag-list-item"><a class="tag-list-link" href="/tags/go/" rel="tag">go</a></li><li class="tag-list-item"><a class="tag-list-link" href="/tags/mpsc/" rel="tag">mpsc</a></li></ul>
            </div>
<div class="toc">
  
</div>
            <div class="content">
                <p>本文实现了一个介于wait-free 和 lock-free的高性能MPSC队列，也就是多生产者但消费者队列。</p>
<p>与使用channel耗时测试：<br>测试配置：1W个生产者，每个生产者写入90次。统计生产者全部写入队列耗时。</p>
<table>
<thead>
<tr>
<th>测试</th>
<th>MPSC队列耗时</th>
<th>channel耗时</th>
</tr>
</thead>
<tbody><tr>
<td>测试1</td>
<td>115.454185ms</td>
<td>965.9469ms</td>
</tr>
<tr>
<td>测试2</td>
<td>97.00209ms</td>
<td>992.595ms</td>
</tr>
<tr>
<td>测试3</td>
<td>109.776455ms</td>
<td>989.837ms</td>
</tr>
<tr>
<td>测试4</td>
<td>90.19232ms</td>
<td>1.3611857s</td>
</tr>
<tr>
<td>测试5</td>
<td>79.12408ms</td>
<td>1.2725355s</td>
</tr>
<tr>
<td>测试6</td>
<td>68.650755ms</td>
<td>1.3981461s</td>
</tr>
<tr>
<td>测试7</td>
<td>77.155775ms</td>
<td>1.2469043s</td>
</tr>
<tr>
<td>测试8</td>
<td>103.899735ms</td>
<td>1.4227174s</td>
</tr>
<tr>
<td>测试9</td>
<td>72.15339ms</td>
<td>1.3351006s</td>
</tr>
<tr>
<td>测试10</td>
<td>129.974485ms</td>
<td>1.0454128s</td>
</tr>
</tbody></table>
<a id="more"></a>
<pre><code class="go">
package main

import (
    &quot;fmt&quot;
    &quot;runtime&quot;
    &quot;sync&quot;
    &quot;sync/atomic&quot;
    &quot;time&quot;
    &quot;unsafe&quot;
)

type TaskNode struct &#123;
    Data interface&#123;&#125; `json:&quot;data&quot;`
    Next *TaskNode   `json:&quot;Next&quot;`
&#125;

var UNCONNECTED *TaskNode = new(TaskNode)

func NewExecutionQueue(_func func(interface&#123;&#125;)) *ExecutionQueue &#123;
    return &amp;ExecutionQueue&#123;
        Head:          nil,
        _execute_func: _func,
        locker:        sync.Mutex&#123;&#125;,
        pool: &amp;sync.Pool&#123;New: func() interface&#123;&#125; &#123;
            return new(TaskNode)
        &#125;&#125;,
    &#125;
&#125;

type ExecutionQueue struct &#123;
    Head          *TaskNode         `json:&quot;Head&quot;`
    _execute_func func(interface&#123;&#125;) `json:&quot;-&quot;` // 消费者函数
    locker        sync.Mutex        `json:&quot;-&quot;`
    pool          *sync.Pool        `json:&quot;-&quot;`
&#125;

func (ex *ExecutionQueue) AddTaskNode(data interface&#123;&#125;) &#123;
    node := ex.pool.Get().(*TaskNode)
    node.Data = data
    node.Next = UNCONNECTED

    preHead := atomic.SwapPointer((*unsafe.Pointer)(unsafe.Pointer(&amp;ex.Head)), unsafe.Pointer(node))

    if preHead != nil &#123;
        node.Next = (*TaskNode)(preHead)
        return
    &#125;

    node.Next = nil
    // 任务不多直接执行，防止线程切换
    ex._execute_func(node.Data)
    if !ex.moreTasks(node) &#123;
        return
    &#125;
    go ex.exectueTasks(node)

&#125;

func (ex *ExecutionQueue) moreTasks(oldNode *TaskNode) bool &#123;

    newHead := oldNode

    if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&amp;ex.Head)), unsafe.Pointer(newHead), nil) &#123;
        return false
    &#125;
    newHead = (*TaskNode)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&amp;ex.Head))))
    var tail *TaskNode
    p := newHead
    for &#123;
        for &#123;
            if p.Next != UNCONNECTED &#123;
                break
            &#125; else &#123;
                runtime.Gosched()
            &#125;
        &#125;
        saved_next := p.Next
        p.Next = tail
        tail = p
        p = saved_next

        if p == oldNode &#123;
            oldNode.Next = tail
            return true
        &#125;
    &#125;
&#125;

func (ex *ExecutionQueue) exectueTasks(taskNode *TaskNode) &#123;
    for &#123;
        tmp := taskNode

        taskNode = taskNode.Next
        tmp.Next = nil
        ex.pool.Put(tmp)
        ex._execute_func(taskNode.Data)

        if taskNode.Next == nil &amp;&amp; !ex.moreTasks(taskNode) &#123;
            return
        &#125;
    &#125;
&#125;

var count int64 = 0

func print(data interface&#123;&#125;) &#123;
    // a := count
    // _ = a
    _ = data.(int) * data.(int)
    atomic.AddInt64(&amp;count, 1)
    // fmt.Println(data.(int))
&#125;
func Test1() &#123;
    var singalexit = sync.WaitGroup&#123;&#125;
    ex := NewExecutionQueue(print)
    start := time.Now()
    var s string
    for k := 0; k &lt; 20; k++ &#123;
        for i := 0; i &lt; 10000; i++ &#123;
            singalexit.Add(1)
            go func(i int, singalexit *sync.WaitGroup) &#123;
                defer singalexit.Done()
                for j := 0; j &lt; 90; j++ &#123;
                    ex.AddTaskNode(i*100 + j)
                &#125;
            &#125;(i, &amp;singalexit)
            _ = s

        &#125;
    &#125;
    singalexit.Wait()
    elapsed := time.Since(start)
    fmt.Println(&quot;该函数执行完成耗时：&quot;, elapsed/20)
    time.Sleep(2 * time.Second)
    fmt.Println(atomic.LoadInt64(&amp;count))
&#125;

func Test2() &#123;
    var singalexit sync.WaitGroup
    data := make(chan int, 2000)
    var count1 int64 = 0
    go func() &#123;
        for &#123;
            &lt;-data
            atomic.AddInt64(&amp;count1, 1)
        &#125;

    &#125;()
    start := time.Now()
    func() &#123;
        for i := 0; i &lt; 10000; i++ &#123;
            singalexit.Add(1)
            go func(i int) &#123;

                defer singalexit.Done()
                for j := 0; j &lt; 90; j++ &#123;
                    data &lt;- (i*100 + j)
                &#125;
            &#125;(i)
        &#125;
    &#125;()
    singalexit.Wait()
    elapsed := time.Since(start)
    fmt.Println(&quot;该函数执行完成耗时：&quot;, elapsed)
    time.Sleep(2 * time.Second)
    fmt.Println(atomic.LoadInt64(&amp;count1))

&#125;

func main() &#123;
    for i := 0; i &lt; 10; i++ &#123;
        count = 0
        Test1()
    &#125;
    for i := 0; i &lt; 10; i++ &#123;
        Test2()
    &#125;

    // Test2()
&#125;
</code></pre>

            </div>
          
           
            <div class="copyright">
                <div class="name">
                    <a>本文作者:</a>
                    <a>Nick Yang</a>
                </div>
                <div class="link">
                    <a>本文链接:</a>
                    <a class="permalink" href="https://stoneqi.github.io/2021/03/04/go-mpsc-md/">https://stoneqi.github.io/2021/03/04/go-mpsc-md/</a>
                </div>
                <div class="license">
                    <a>版权声明:</a>
                    <a>本博客所有文章除特别声明外，均采用许可协议 CC BY-NC-SA 3.0 CN。转载请注明出处！</a>
                </div>
            </div>
            

          


</article>


<div class="tip">
<button class="tip-btn">
    打赏
</button>
<div class="tip-img">
<ul>
    
<li>
    <img src="/img/aipay.jpg"></img>
</li>

 
<li>
    <img src="/img/wechatpay.png"></img>
</li>

</ul>
</div>
</div>

<div class="more">

    <div class="prev">
   <a href="/2021/03/05/tr-golang-complier-part-1-md/">  【翻译】Go 编译器理解: 新增一个关键字语句-Part 1</a>
    </div>

<div></div>

    <div class="next">
    <a href="/2021/03/04/go-nil-type-md/"> Go nil详解 </a>
    </div>
    
</div>

<div class="bdsharebuttonbox">
<a href="#" class="bds_weixin fa fa-weixin" data-cmd="weixin" title="分享到微信" style="color:#1cbd8f">
</a>
<a href="#" class="bds_tsina fa fa-weibo" data-cmd="tsina" title="分享到新浪微博" style="color:#ff6363">
</a>
<a href="#" class="bds_twi fa fa-twitter" data-cmd="twi" title="分享到Twitter" style="color:#00A7EB">
</a>
<a href="#" class="bds_fbook fa fa-facebook" data-cmd="fbook" title="分享到Facebook" style="color:#00A7EB">
</a>
</div>
<script>
window._bd_share_config={
    "common":{"bdSnsKey":{},"bdText":"","bdMini":"2","bdMiniList":false,"bdPic":"","bdStyle":"1","bdSize":"16"},
    "share":{}};with(document)0[(getElementsByTagName('head')[0]||body).appendChild(createElement('script')).src='../../../../../static/api/js/share.js?v=89860593.js?cdnversion='+~(-new Date()/36e5)];
</script>
 
<!-- <div id="comments"></div> -->
<div id="gitalk-container"></div>

<script>
  //   var gitment = new Gitment({
  //   id: "Thu Mar 04 2021 11:06:13 GMT+0800",
  //   owner: "StoneQi",
  //   repo: "stoneqi.github.io",
  //   oauth: {
  //     client_id:"d9f4e8907ec2566c2307",
  //     client_secret: "2390fe6084c9227566e82de1755e3c3de4a13592",
  //   }
  // })
  // gitment.render('comments')

  const gitalk = new Gitalk({
  clientID: "d9f4e8907ec2566c2307",
  clientSecret: "2390fe6084c9227566e82de1755e3c3de4a13592",
  repo: "stoneqi.github.io",      // The repository of store comments,
  owner: "StoneQi",
  admin: ["StoneQi"],
  language: "zh-CN",
  id: "Thu Mar 04 2021 11:06:13 GMT+0800",      // Ensure uniqueness and length less than 50
  distractionFreeMode: false  // Facebook-like distraction free mode
})

gitalk.render('gitalk-container')
</script>
    </main>
    <a class="not-found">not found!</a>
    <div class="search-items">
    </div>
    <a href="#header" id="top" style="display:none">
        <i class="fa fa-sort-asc fa-2x"></i>
    </a>
    <footer class="footer">
    <div class="footer-copyright">©️2017
    <a href="//github.com/Vevlins/toki" class="link" target="_blank">Toki</a>  by Vevlins
    </div>    
</footer>

    
<script src="/js/jquery.js"></script>

    
<script src="/js/toki.js"></script>
  
  <script>hljs.initHighlightingOnLoad();</script>
</body>
</html>